package com.yanggu.flink.datastream_api.process_function

import com.yanggu.flink.datastream_api.pojo.SensorReading
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * ProcessFunction flink的底层api, 可以访问到当前数据的EventTime、Watermark、Timer等
 * 所有的ProcessFunction都是RichFunction的子类, 可以访问open、close、getRuntimeContext等生命周期方法
 * 除此之外还提供processElement和onTimer方法
 * 同时还能够进行侧输出流
 * 状态KeyedState和Timer定时器只能在KeyedProcessFunction中使用
 */
object ProcessFunctionDemo01 {

  /**
   * 监控温度传感器的温度值，如果温度值在10秒钟之内(processing time)连续上升，则报警
   */
  def main(args: Array[String]): Unit = {

    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    env.socketTextStream("localhost", 9000)
      .map(data => {
        val strings = data.split(",")
        SensorReading(strings(0), strings(1).toLong, strings(2).toDouble)
      })
      .keyBy(_.id)
      .process(new TempIncreaseAlertFunction(10 * 1000L))
      .print("告警输出")

    env.execute("测试定时器功能")

  }

}

/**
 * 第一个泛型 Key的类型
 * 第二个泛型 Input的类型
 * 第三个泛型 Output的类型
 * 监控温度传感器的温度值，如果温度值在10秒钟之内(processing time)连续上升，则报警
 *
 * @param continuousRiseTime 连续上升的时间阈值, 单位毫秒值
 * @tparam KEY    String key的类型
 * @tparam Input  SensorReading 输入的数据类型
 * @tparam Output String 输出的数据类型
 */
class TempIncreaseAlertFunction(continuousRiseTime: Long) extends KeyedProcessFunction[String, SensorReading, String] {

  //上一个温度的状态
  private lazy val lastTemperatureValueState = getRuntimeContext.getState(
    new ValueStateDescriptor[Double]("lastTemperatureValueState", classOf[Double]))

  //定时器的时间戳的状态
  private lazy val timerTimeStampValueState = getRuntimeContext.getState(
    new ValueStateDescriptor[Long]("timerTimeStampValueState", classOf[Long]))

  /**
   * 分区中的流中的每一个元素都会调用这个方法
   *
   * @param data
   * @param ctx
   * @param out
   */
  override def processElement(data: SensorReading,
                              ctx: KeyedProcessFunction[String, SensorReading, String]#Context,
                              out: Collector[String]): Unit = {
    val key = ctx.getCurrentKey
    val long = ctx.timestamp()
    val service = ctx.timerService()
    val currentProcessingTime = service.currentProcessingTime()
    val currentMark = service.currentWatermark()
    //打印key、当前时间戳、处理时间、watermark等
    println(s"当前的key: $key, 当前的时间戳: $long, 当前的处理时间: $currentProcessingTime, 当前的水位: $currentMark")


    val lastTemperature = lastTemperatureValueState.value()
    val timerTimeStamp = timerTimeStampValueState.value()
    val currentTemperature = data.temperature

    //更新上一次温度的状态, 即更新为本次的温度
    lastTemperatureValueState.update(currentTemperature)

    //如果温度下降或是第一个温度值, 直接删除定时器
    if (currentTemperature <= lastTemperature || lastTemperature == 0.0D) {
      service.deleteProcessingTimeTimer(timerTimeStamp)
      timerTimeStampValueState.clear()
      //如果本次温度大于上次温度并且没有设置定时器, 设置定时器10s后执行
    } else if (currentTemperature > lastTemperature && timerTimeStamp == 0) {
      val registerTimeStamp = currentProcessingTime + continuousRiseTime
      //这里设置的是执行时间定时器
      service.registerProcessingTimeTimer(registerTimeStamp)
      //更新定时器时间戳的状态
      timerTimeStampValueState.update(registerTimeStamp)
    }
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[String, SensorReading, String]#OnTimerContext,
                       out: Collector[String]): Unit = {
    out.collect(s"传感器: ${ctx.getCurrentKey}, 在${continuousRiseTime / 1000}秒内温度连续上升")
    //清除状态
    timerTimeStampValueState.clear()
  }

}
